package com.niit.streaming

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object Spark_stream_Kafka {

  def main(args: Array[String]): Unit = {
                                            //本地模式 *：使用CPU核数作为分区数
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStream")
    val ssc = new StreamingContext(sparkConf,Seconds(3))
    ssc.sparkContext.setLogLevel("ERROR")

    val kafkaPara = Map[String,Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "node1:9092",
      ConsumerConfig.GROUP_ID_CONFIG ->"SP_KF",
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer].getName,
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer].getName
    )

    //读取Kafka数据创建 DStream
    val kafkaDataDS: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      ssc,
      LocationStrategies.PreferConsistent, //官方推荐本地策略
      ConsumerStrategies.Subscribe[String, String](Set("BD1"), kafkaPara)
    )
    //打印读取到的数据
    val infoDS: DStream[String] = kafkaDataDS.map(record => {
      val topic = record.topic();
      val partition = record.partition()
      val offset = record.offset()
      val value = record.value()
      val info = s"主题${topic},分区${partition},偏移量${offset},值${value}"
      info
    })
    infoDS.print()


    ssc.start()
    ssc.awaitTermination()
  }

}
